/*================================================================*
        Copyright (C) 2021 All rights reserved, www.hsby.link.
      	文件名称：echo_server.cc
      	创 建 者：汉森伯逸
      	创建日期：2022/1/16
 *================================================================*/

#include <sys/epoll.h>
#include <iostream>
#include <fstream>
#include <cstring>
#include <assert.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <sys/syscall.h>
#include <sys/time.h>
#include <fcntl.h>
#include <netinet/tcp.h>
#include <thread>
#include <vector>
#include <semaphore.h>
#include "util/nolock_queue.h"

using namespace std;

std::ofstream logf("log", std::ofstream::out);

#define STRVAR(var) (std::string(#var) + " = " + std::to_string(var))
#define STRERR ("errno = " + std::to_string(errno) + ", strerr = " + strerror(errno))
//#define LOG1(msg) logf << __LINE__ << "|" << msg << endl;
#define LOG1(msg) cout << __LINE__ << "|" << msg << endl;
//#define LOG1(msg)
//#define LOG2(msg) logf << __LINE__ << "|" << msg << endl;
//#define LOG2(msg) cout << __LINE__ << "|" << msg << endl;
#define LOG2(msg)

struct TimeStamp {
    TimeStamp() {
        Point();
    }

    int64_t Point() {
//        struct timeval tv;
//        gettimeofday(&tv, nullptr);
//        return last_time_stamp = tv.tv_sec * 1000 * 1000 + tv.tv_usec;
        return 0;
    }

    int64_t GetInterval(const string& prefix) {
//        auto tmp = last_time_stamp;
//        int64_t now_ms = Point();
//        int64_t res = now_ms - tmp;
//        LOG2(prefix << "-interval = " << res << "us");
        return 0;
    }

    int64_t last_time_stamp = 0;
};

// 有锁消息队列
template<typename T>
struct MessageQueue {
	MessageQueue(int size) {
		sem_init(&sem_capa, 0, size);
		sem_init(&sem_data, 0, 0);
	}

	void Push(const T& msg) {
		sem_wait(&sem_capa);
		data.Push(msg);
        size++;
		sem_post(&sem_data);
	}

	T Pull() {
		sem_wait(&sem_data);
		auto msg = data.Pull();
		// 获取成功则消费空位, 否则返还资料
        if (msg) {
            size--;
            sem_post(&sem_capa);
        }else {
            sem_post(&sem_data);
        }
		return msg;
	}

	std::atomic<int> size = {0};
	sem_t sem_capa;
	sem_t sem_data;
	hsby::NoLockQueue<T> data;
};

const int MSG_LIST_SIZE = 1024 * 4;
MessageQueue<int> read_clients(MSG_LIST_SIZE);
MessageQueue<int> write_clients(MSG_LIST_SIZE);

const string rsp =    "HTTP/1.1 200 OK\r\nContent-length:1\r\n\r\na\r\n";
//const string rsp =    "HTTP/1.1 200 OK\r\n\r\n";
int listen_sock = -1;
int ep = -1;
thread_local TimeStamp time_stamp; //时间戳工具

void ReadHandler() {
    epoll_event epev_tmp;
	string req;
	req.resize(4096);
	while(true) {

		auto client = read_clients.Pull();
		if (!client) {
            continue;
		}
        LOG1("read clients count = " << read_clients.size);
        time_stamp.GetInterval("ReqHandler|1.PullClient");

        int read_len = 0;
		while (true) {
            int ret = read(client, &req[0], req.size());
            if (ret <= -1 && errno == EAGAIN) {
                time_stamp.GetInterval("ReqHandler|2.1 Read(-1)");
                LOG1("read return -1 && EGAIN, " << STRVAR(client) << ", " << STRERR);
                break;
            }
            if (ret == 0) {
                if (read_len == 0) {
                    // 关闭连接
                    epoll_ctl(ep, EPOLL_CTL_DEL, client, nullptr);
                    close(client);
                }
                break;
            }
            read_len += ret;
		}
        time_stamp.GetInterval("ReqHandler|2 ReadComplete");
        LOG1(STRVAR(client) << ", " << STRVAR(read_len));

        // 监听可写
        memset(&epev_tmp, 0, sizeof(epev_tmp));
        epev_tmp.data.fd = client;
        epev_tmp.events |= EPOLLOUT;
        epev_tmp.events |= EPOLLET;
        epoll_ctl(ep, EPOLL_CTL_MOD, client, &epev_tmp);
	}
}

void WriteHandler() {
    epoll_event epev_tmp;
    while(true) {
        auto client = write_clients.Pull();
        if (!client) {
            continue;
        }
        LOG1("write clients count = " << write_clients.size);
        int write_len = 0;
        while (write_len < (int)rsp.size()) {
            int ret = write(client, &rsp[write_len], rsp.size() - write_len);
            if (ret <= -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
                time_stamp.GetInterval("ReqHandler|3.1 Write(-1)");
                LOG1("write contineue");
                break;
            }
            if (ret == 0) {
                break;
            }
            write_len += ret;
        }
        time_stamp.GetInterval("ReqHandler|3.WriteComplete");
        LOG1(STRVAR(client) << ", " << STRVAR(write_len));

        // 监听可读
        memset(&epev_tmp, 0, sizeof(epev_tmp));
        epev_tmp.data.fd = client;
        epev_tmp.events |= EPOLLIN;
        epev_tmp.events |= EPOLLET;
        epoll_ctl(ep, EPOLL_CTL_MOD, client, &epev_tmp);
    }
}

void Loop() {
    epoll_event epev_tmp;
    while (true) {
        // epoll等待
        epoll_event ep_evs[1024 * 10];
        int ret = epoll_wait(ep, ep_evs, sizeof(ep_evs) / sizeof(epoll_event), -1);
        time_stamp.GetInterval("Accept|1.epoll_wait");
        LOG1("accept epoll_wait return " << ret);

        if (ret < 0) {
            continue;
        }

        // 处理请求
        for (int i = 0; i < ret; i++) {
            auto& epev = ep_evs[i];
            if (epev.data.fd == listen_sock) {      // 如果是监听套接字，接收新请求
                if (epev.events & EPOLLIN) {
                    LOG1( STRVAR(epev.data.fd) << " EPOLLIN");
                    int client = accept(epev.data.fd, nullptr, nullptr);
                    time_stamp.GetInterval("Accept|2.1.accept");
                    if (client <= 0) {
                        continue;
                    }
                    LOG1("accept client = " << client);

                    // 设置非阻塞
                    int flag = fcntl(client, F_GETFL, 0);
                    fcntl(client, F_SETFL, flag | O_NONBLOCK);
                    time_stamp.GetInterval("Accept|2.2fcntl");
                    // 进制nagle算法
                    int val = 1;
                    setsockopt(client, IPPROTO_TCP, TCP_NODELAY, &val, sizeof(val));

                    // 加入epoll
                    memset(&epev_tmp, 0, sizeof(epev_tmp));
                    epev_tmp.data.fd = client;
                    epev_tmp.events |= EPOLLIN;
                    epev_tmp.events |= EPOLLET;
                    epoll_ctl(ep, EPOLL_CTL_ADD, client, &epev_tmp);
                    time_stamp.GetInterval("Accept|2.3.epoll_ctl");
                }
            }else {
                int client = epev.data.fd;
                // 对端是否出错或关闭
                if (epev.events & EPOLLHUP || epev.events & EPOLLRDHUP || epev.events & EPOLLERR) {
                    epoll_ctl(ep, EPOLL_CTL_DEL, client, nullptr);
                    close(client);
                    continue;
                }
                if (epev.events & EPOLLIN){   // 如果是请求到达, 抛出可读请求
                    LOG1( STRVAR(client) << " EPOLLIN");

                    // 投递读请求
                    read_clients.Push(client);
                    time_stamp.GetInterval("Accept|3.2.PushClients");
                }else if (epev.events & EPOLLOUT) {
                    LOG1( STRVAR(client) << " EPOLLOUT");
                    // 投递写请求
                    write_clients.Push(client);
                    time_stamp.GetInterval("Accept|3.1.PushWriteClients");
                }
            }
        }
    }

}

int main(int argc, char** args) {
    if (argc < 4) {
        cout << "Usage: bin LoopCount ReadHandlerCount WriteHandlerCount" << endl;
        return -1;
    }
    int loop_handler_count = atoi(args[1]);
    int read_handler_count = atoi(args[2]);
    int write_handler_count = atoi(args[3]);

    // 初始化epoll
    ep = epoll_create(1024 * 10);
    assert(ep > 0);

    // 设置监听地址
    sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    inet_pton(AF_INET, "0.0.0.0", &addr.sin_addr.s_addr);
    addr.sin_port = htons(80);

    // 申请套接字
    listen_sock = socket(AF_INET, SOCK_STREAM, 0);
    assert(listen_sock > 0);

	int res = 1;
	int ret = setsockopt(listen_sock, SOL_SOCKET, SO_REUSEADDR, &res, sizeof(res));
	assert(ret == 0);
	ret = setsockopt(listen_sock, SOL_SOCKET, SO_REUSEPORT, &res, sizeof(res));
	assert(ret == 0);

    ret = setsockopt(listen_sock, IPPROTO_TCP, TCP_NODELAY, &res, sizeof(res));
    assert(ret == 0);
    int flag = fcntl(listen_sock, F_SETFL, 0);
    fcntl(listen_sock, F_SETFL, flag | O_NONBLOCK);

    // 绑定
    ret = bind(listen_sock, (sockaddr*)&addr, sizeof(addr));
    assert(ret == 0);

    // 设置监听
    ret = listen(listen_sock, 1024 * 40);
    assert(ret == 0);

    // 设置epoll
    epoll_event epev_tmp;
    memset(&epev_tmp, 0, sizeof(epev_tmp));
    epev_tmp.data.fd = listen_sock;
    epev_tmp.events = EPOLLIN;
    epoll_ctl(ep, EPOLL_CTL_ADD, listen_sock, &epev_tmp);

    // 开启线程
    std::vector<thread> ts;
    for (int i = 0; i < loop_handler_count; ++i) {
        ts.push_back(thread(Loop));
    }
    for (int i = 0; i < read_handler_count; ++i) {
        ts.push_back(thread(ReadHandler));
    }
    for (int i = 0; i < write_handler_count; ++i) {
        ts.push_back(thread(WriteHandler));
    }

    for (auto& t : ts) {
        t.join();
    }

	// 关闭epoll
    close(ep);
	// 关闭监听
	close(listen_sock);
    return 0;
}
